package com.cong.queue;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

class MySource{
    //开启生产消费
    private volatile boolean flag =true;

    private  AtomicInteger atomicInteger = new AtomicInteger();
    /**
     * 阻塞队列
     */
    BlockingQueue<String> blockingQueue = null;

    /**
     *
     * @param blockingQueue
     */
    public MySource(BlockingQueue<String> blockingQueue) {
        this.blockingQueue = blockingQueue;
        //获取传入的队列类型
        System.out.println(blockingQueue.getClass().getName());
    }

    /**
     *
     */
    public void MyProduct() throws InterruptedException {

        String data =null;
        boolean retValue;

        while (flag){
            data=String.valueOf( atomicInteger.incrementAndGet());
            retValue = blockingQueue.offer(data,2L,TimeUnit.SECONDS);
            if (retValue){
                System.out.println(Thread.currentThread().getName()+"\toffer"+data+"success");
            }else {
                System.out.println(Thread.currentThread().getName()+"\toffer"+data+"false");
            }
            TimeUnit.SECONDS.sleep(1);
        }
        System.out.println(Thread.currentThread().getName()+"\tboss stop flag=false");
    }

    public void MyConsumer() throws InterruptedException {

        String result =null;
        while (flag){
            result =blockingQueue.poll(2L, TimeUnit.SECONDS);
            if (null==result || result.equalsIgnoreCase("")){
                flag =false;
                System.out.println();
                System.out.println(Thread.currentThread().getName()+"\t consumer time out 2s");
                //退出
                return;
            }
            System.out.println(Thread.currentThread().getName()+"\t consumer"+result+"success");
        }
    }

    public void stop(){
        flag=false;
    }

}
/**
 * @author cong
 * <p>
 * created on 2021/3/31 上午11:52
 */
public class ProdConsumerQueueDemo {

    public static void main(String[] args) {
        MySource mySource = new MySource(new ArrayBlockingQueue<>(10));
        new Thread(()->{
            try {
                System.out.println("start product");
                mySource.MyProduct();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"product").start();

        new Thread(()->{
            try {
                System.out.println("consumer start");
                mySource.MyProduct();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"consumer").start();

        try {
            TimeUnit.SECONDS.sleep(5L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("5s end");
        mySource.stop();
    }
}
